package com.sk.rpcAssist.dataManager;


import com.sk.rpcAssist.domain.ArgumentConfig;
import com.sk.rpcAssist.domain.BasicConfig;
import com.sk.rpcAssist.domain.MethodConfig;
import com.sk.rpcAssist.domain.ServiceType;
import com.sk.rpcAssist.policy.ServiceBeanDefinitionRegistry;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.core.io.ClassPathResource;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.*;

import static java.util.stream.Collectors.toList;

/**
 * description：使用curator作为zk的client连接工具
 * [功能点] 对znode的crud、初始化client（包含重试策略）
 *
 * @author 孙文斌
 * @Date: 2021/1/31 13:07
 */

@Data
@Slf4j
public class ConsumerCuratorClient {

    private String connectUrls;
    private String retryTime;
    private String connectionTimeOut;
    private String sessionTimeOut;
    private String baseSleepTime;
    private String port;
    private String host;

    private final String ZK_CURATOR_CONNECTURLS = "zk.curator.connectUrls";
    private final String ZK_CURATOR_RETRYTIME = "zk.curator.retryTime";
    private final String ZK_CURATOR_CONNECTIONTIMEOUT = "zk.curator.connectionTimeOut";
    private final String ZK_CURATOR_SESSIONTIMEOUT = "zk.curator.sessionTimeOut";
    private final String ZK_CURATOR_BASESLEEPTIME = "zk.curator.baseSleepTime";
    private final String PORT = "server.port";
    private ApplicationContext applicationContext;

    private CuratorFramework client;

    private ConsumerDataManager consumerDataManager;

    private List<String> preProvidePaths = new ArrayList();

    public ConsumerCuratorClient(ConsumerDataManager defProMang) {
        YamlPropertiesFactoryBean yamlMapFactoryBean = new YamlPropertiesFactoryBean(); //可以加载多个yml文件
        yamlMapFactoryBean.setResources(new ClassPathResource("application.yml"));
        Properties properties = yamlMapFactoryBean.getObject(); //获取yml里的参数
        this.connectUrls = properties.getProperty(ZK_CURATOR_CONNECTURLS);
        this.retryTime = properties.getProperty(ZK_CURATOR_RETRYTIME);
        this.connectionTimeOut = properties.getProperty(ZK_CURATOR_CONNECTIONTIMEOUT);
        this.sessionTimeOut = properties.getProperty(ZK_CURATOR_SESSIONTIMEOUT);
        this.baseSleepTime = properties.getProperty(ZK_CURATOR_BASESLEEPTIME);
        this.port = properties.getProperty(PORT);
        this.consumerDataManager = defProMang;
        //获取当前host
        this.host = getIpAddress();
        //todo:参数校验
        //启动curator
        init();
        //注册自己的路径
        doRegister();
        //拉取所有监听的节点
        doListener();
    }

    private void doListener() {
        if (CollectionUtils.isEmpty(preProvidePaths)) {
            return;
        }
        for (String path : preProvidePaths) {
            listenerChild(path);
        }
    }

    private void doRegister() {
        if (null == client) {
            throw new RuntimeException("register consumer info fail,there is not connect to zkServer...");
        }
        //待注册接口
        Set<Class<?>> consumerBeanClazzs = ServiceBeanDefinitionRegistry.consumerBeanClazzs;

        List<BasicConfig> consumerCollect = consumerBeanClazzs.stream().map(bo -> {
            BasicConfig config = class2BasicConfig(bo, ServiceType.CONSUMER);
            return config;
        }).collect(toList());
        for (BasicConfig config : consumerCollect) {
            //注册consumer
            register(config);
        }

//        List<BasicConfig> provideCollect = consumerBeanClazzs.stream().map(bo -> {
//            BasicConfig config = class2BasicConfig(bo, ServiceType.PROVIDER);
//            config.setPort(null);
//            config.setHost(null);
//            return config;
//        }).collect(toList());

        //创建待监听的provide节点【没必要创建，在使用下面缓存监听模式的时候，会自动创建需要监听的路径，而且都是持久节点】
//        for (BasicConfig config : provideCollect) {
//            //预注册provide
//            preRegister(config);
//        }
    }

    private BasicConfig class2BasicConfig(Class<?> bo, ServiceType serviceType) {
        BasicConfig config = null;
        if (bo != null) {
            config = new BasicConfig();
            config.setHost(this.host);
            config.setPort(Integer.valueOf(this.port));
            config.setInterfaceClass(bo);
            config.setInterfaceName(bo.getName());
            config.setServiceType(serviceType.CONSUMER);
            Method[] methodArr = bo.getMethods();
            List<MethodConfig> methods = new ArrayList<>();
            for (Method method : methodArr) {
                MethodConfig methodConfig = new MethodConfig();
                methodConfig.setMethodName(method.getName());
                methodConfig.setReturnType(method.getReturnType());
                methodConfig.setReturnFlag(!(method.getReturnType() == Void.TYPE));
                methodConfig.setParameterTypes(method.getParameterTypes());
                Type[] parameterTypes = method.getGenericParameterTypes();
                List<ArgumentConfig> argumentConfigs = new ArrayList<>();
                int i = 0;
                for (Type type : parameterTypes) {
                    ArgumentConfig argumentConfig = new ArgumentConfig();
                    argumentConfig.setIndex(i++);
                    argumentConfig.setType(type.getTypeName());
                    argumentConfigs.add(argumentConfig);
                }
                methodConfig.setArgumentConfigs(argumentConfigs);
                methods.add(methodConfig);
            }
            config.setMethods(methods);
        }
        return config;
    }

    private static String getIpAddress() {
        InetAddress address = null;
        try {
            address = InetAddress.getLocalHost();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        return address.getHostAddress();
    }


    private CuratorFramework init() {
        try {
            client = CuratorFrameworkFactory.builder()
                    .connectString(connectUrls)
                    .connectionTimeoutMs(Integer.valueOf(connectionTimeOut))
                    .sessionTimeoutMs(Integer.valueOf(sessionTimeOut))
                    .retryPolicy(new ExponentialBackoffRetry(Integer.valueOf(baseSleepTime), Integer.valueOf(retryTime)))
                    .build();
            client.start();
        } catch (Exception e) {
            e.printStackTrace();
            log.error("init zookeeper client fail,param is " + this.toString());
        }
        return client;
    }

    /**
     * @param
     * @Description: 关闭连接
     * @Date: 2021/1/31 13:07
     */
    public void closeConnection() {
        if (null != client) {
            client.close();
        }
    }

    /**
     * @param
     * @Description: 使用缓存监听模式(Path Cache),【此模式会自动创建需要监听的路径】
     * @Date: 2021/2/05 13:07
     */
    public void listenerChild(String path) {
        //构造一个缓存实例PathChildrenCache
        /*
         * description：
         *
         * @param [path]
         * @param [path]
         * @param [cacheData]是否把节点内容缓存起来
         * @param [dataIsCompressed]是否对节点数据进行压缩
         * @param [executorService]异步处理监听事件的执行器
         * @date: 2021/2/5
         **/
        try {
            PathChildrenCache pathChildrenCache = new PathChildrenCache(this.client, path, true);//只能监听子节点，不能递归监听子节点下的子节点
            //构造一个子节点缓存监听器PathChildrenCacheListener实例
            PathChildrenCacheListener listener = new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                    try {
                        List<ChildData> childDataList = pathChildrenCache.getCurrentData();
                        consumerDataManager.synData(childDataList, event);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            };
            pathChildrenCache.getListenable().addListener(listener);
            //NORMAL——异步初始化cache
            //BUILD_INITIAL_CACHE——同步初始化cache
            //POST_INITIALIZED_EVENT——异步初始化cache，并触发完成事件
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
            //首次拉取provide信息
            //todo:暂时不考虑更新和首次拉取的次序问题
//            defaultProvideInfoManager.pullProvideInfo(pathChildrenCache.getCurrentData());
        } catch (Exception e) {
            e.printStackTrace();
            log.error("使用缓存监听模式(Path Cache)对consumer监听失败，e+" + e);
        }
    }

    public void register(BasicConfig config) {
        String interfacePath = "/" + config.getInterfaceName();
        try {
            if (this.client.checkExists().forPath(interfacePath) == null) {
                // 创建 服务的永久节点
                this.client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .forPath(interfacePath);
            }
            config.getMethods().forEach(method -> {
                try {
                    String methodPath = null;
                    ServiceType serviceType = config.getServiceType();
                    String address = getServiceAddress(config);
                    //key=/interfaceName/provider/method+parameterTypes类型字符串的hashcode值
                    Class<?>[] parameterTypes = method.getParameterTypes();
                    StringBuilder sb = new StringBuilder();
                    for (Class<?> parameterType : parameterTypes) {
                        sb.append(parameterType.getName());
                    }
                    int hashCode = sb.toString().hashCode();
                    if (serviceType == ServiceType.CONSUMER) {
                        // 服务提供方，需要暴露自身的ip、port信息，而消费端也暴露，供后续治理使用
                        //consumer供监听使用的path
                        preProvidePaths.add(String.format("%s/%s/%s", interfacePath, ServiceType.PROVIDER.getType(), method.getMethodName()+hashCode));
                    }
                    methodPath = String.format("%s/%s/%s/%s", interfacePath, serviceType.getType(), method.getMethodName()+hashCode, address);
                    log.info("zk path: [" + methodPath + "]");
                    this.client.create()
                            .creatingParentsIfNeeded()
                            .withMode(CreateMode.EPHEMERAL)//最末尾的是临时节点
                            .forPath(methodPath, "0".getBytes());
                    // 创建临时节点，节点包含了服务提供段的信息
                } catch (Exception e) {
                    e.getMessage();
                }
            });
        } catch (Exception e) {
            e.getMessage();
        }
    }

//    public InetSocketAddress discovery(RpcRequest request, ServiceType nodeType) {
//        String path = String.format("/%s/%s/%s", request.getClassName(), nodeType.getType(), request.getMethodName());
//        try {
//            List<String> addressList = this.client.getChildren().forPath(path);
//            // 采用负载均衡的方式获取服务提供方信息,不过并没有添加watcher监听模式
////            String address = loadBalance.balance(addressList);
//            String address = null;
//            if (address == null) {
//                return null;
//            }
//            return parseAddress(address);
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//        return null;
//    }

    private String getServiceAddress(BasicConfig config) {
        String hostInfo = "";
        if (!StringUtils.isEmpty(config.getHost()) || config.getPort() != null) {
            hostInfo = new StringBuilder()
                    .append(config.getHost())
                    .append(":")
                    .append(config.getPort())
                    .toString();
        }
        return hostInfo;
    }

    private InetSocketAddress parseAddress(String address) {
        String[] result = address.split(":");
        return new InetSocketAddress(result[0], Integer.valueOf(result[1]));
    }
}
